#include "gtest/gtest.h"

#include <iostream>
#include <atomic>
#include <thread>

#include "SimpleAppender.h"
#include "SimpleSequenceReader.h"
#include "Topic.h"
#include "FileSize.h"
#include "recovery/Block.h"

using recovery::Block;


TEST(SequenceReaderTest, threads){
    auto tp = std::make_unique<Topic>("record", ".");
    std::atomic<int> status = 0; //0: init, 1: start, 2: stop
    std::atomic<size_t> total = -1;

    auto produce = [&] () {
        auto ap = std::make_unique<SimpleAppender>(*tp);
        ASSERT_TRUE(ap->open());
        status = 1;
        for (int i = 1; i <= 1000; i++) {
            ap->append(std::to_string(i));
            ap->commit();
        }
        ASSERT_TRUE(ap->close());
        total = tp->fileSize()->size();
        status = 2;
    };
    std::thread producer(produce);

    auto consume = [&]() {
        auto seqr = std::make_unique<SimpleSequenceReader>(*tp);
        while (status != 1) {}  //for file not exist, then seqr->open failed
		ASSERT_TRUE(seqr->open());
		auto block = std::make_unique<Block>();
        
	    std::string read;
	    size_t readBytes = 0;
        while (!(status == 2 && read.size() == total)) {
            int ret = seqr->polling(20);
			if (ret > 0){ 
				auto s = seqr->read(block->addr(), (size_t)ret < block->bytes()? ret : block->bytes(), readBytes);
				if(s){
                    read += std::string(reinterpret_cast<const char *>(block->addr()), readBytes);
                    if ((size_t)ret < block->bytes()) {
                        ASSERT_EQ(read.substr(read.size() - 7), "commit;");
                    }
				}
				else{
					std::cerr << std::this_thread::get_id() << ": read error" << std::endl;	
					break;
				}
			}
            else if (ret < 0) {
                std::cerr << std::this_thread::get_id() << ": polling got an error" << std::endl;
                break;
            }
        }
        ASSERT_TRUE(seqr->close());
    };
    std::thread consumer1(consume);
    std::thread consumer2(consume);
    std::thread consumer3(consume);

    producer.join();
    consumer1.join();
    consumer2.join();
    consumer3.join();
}
